package com.lg.customrxjava;

public  class Observable< T> {

    ObservableOnSubscribe source;

    private Observable(ObservableOnSubscribe<T> source) {
        this.source = source;
    }


    public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
            return new Observable<T>(source) ;
    }

    public static <T> Observable<T> just(final T... t) {
        ObservableOnSubscribe onSubscribe= new ObservableOnSubscribe<T>() {
            @Override
            public void subscribe(Observer<T> observableEmitter) {
//                observableEmitter.onNext(t);
//                observableEmitter.onComplete();
                for (T t1:t) {
                    observableEmitter.onNext(t1);
                }
                 observableEmitter.onComplete();

            }
        };

        return new Observable<T>(onSubscribe) ;
    }

    //? super 可写 ｜？ extends 可读
    public  <R> Observable<R> map(Function<? super T, ? extends R>  function){
          //  ObservableMap map = new ObservableMap(source,function);
        ObservableMap<T, R> observableMap = new ObservableMap(source, function); // source 控制权

        return new Observable<R>(observableMap); // observableMap是source的实现类
        //return new Observable(map) ;
    }


//    public  <R> Observable<R> map(Function<T,  R>  function){
//        //  ObservableMap map = new ObservableMap(source,function);
//        ObservableMap<T, R> observableMap = new ObservableMap(source, function); // source 控制权
//
//        return new Observable<R>(observableMap); // observableMap是source的实现类
//        //return new Observable(map) ;
//    }

    public void subscribe(Observer<T> observer) {
        observer.onSubscribe();

            source.subscribe( observer);


    }


    private class CreateEmitter<T> implements Observer<T>{
        private static final long serialVersionUID = -3434801548987643227L;

        final Observer<? super T> observer;

        CreateEmitter(Observer<? super T> observer) {
            this.observer = observer;
        }

        @Override
       public void onNext( T value){
            observer.onNext(value);
        };

        @Override
        public void onError( Throwable error){
            observer.onError(error);
        }
        @Override
        public void onComplete(){
            observer.onComplete();
        }

        @Override
        public void onSubscribe() {
        }

        ;
    }
}
